/*
 * SpringBoot WebFlux异步编程示例 - Java响应式Web框架详解
 * 
 * 本文件展示SpringBoot WebFlux的响应式编程特性，包括响应式路由、异步数据处理、
 * 流式响应、背压处理等高级功能。
 * 
 * 与FastAPI异步对比：
 * 1. SpringBoot WebFlux使用Reactor框架，FastAPI使用asyncio
 * 2. Java响应式编程学习曲线更陡峭，Python async/await更直观
 * 3. WebFlux背压处理更完善，FastAPI需要手动控制
 * 4. WebFlux生态更成熟，FastAPI异步生态更轻量
 * 
 * 作者: Assistant
 * 日期: 2024
 */

package com.example.webflux;

// ============================================================================
// 1. 依赖导入
// ============================================================================

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.*;
import org.springframework.http.ResponseEntity;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.stereotype.Service;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ControllerAdvice;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import reactor.util.retry.Retry;

import javax.validation.Valid;
import javax.validation.constraints.*;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonProperty;

import java.time.LocalDateTime;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import lombok.Data;
import lombok.Builder;
import lombok.NoArgsConstructor;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

// ============================================================================
// 2. 响应式数据模型
// ============================================================================

/**
 * 异步任务请求模型
 * 相当于Python的TaskRequest
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
class ReactiveTaskRequest {
    
    @NotBlank(message = "任务名称不能为空")
    private String name;
    
    @Min(value = 1, message = "任务持续时间不能小于1秒")
    @Max(value = 60, message = "任务持续时间不能大于60秒")
    private Integer duration;
    
    @NotBlank(message = "任务类型不能为空")
    private String taskType;
    
    private Map<String, Object> params = new HashMap<>();
}

/**
 * 异步任务响应模型
 * 相当于Python的TaskResponse
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
class ReactiveTaskResponse {
    
    private String taskId;
    
    private String name;
    
    private String status;
    
    private Double progress;
    
    private Object result;
    
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss")
    private LocalDateTime createdAt;
    
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss")
    private LocalDateTime startedAt;
    
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss")
    private LocalDateTime completedAt;
    
    private String error;
}

/**
 * 响应式API响应格式
 * 相当于Python的ApiResponse
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
class ReactiveApiResponse<T> {
    
    private boolean success;
    
    private String message;
    
    private T data;
    
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss")
    private LocalDateTime timestamp = LocalDateTime.now();
    
    // 静态工厂方法
    public static <T> ReactiveApiResponse<T> success(String message, T data) {
        return ReactiveApiResponse.<T>builder()
            .success(true)
            .message(message)
            .data(data)
            .timestamp(LocalDateTime.now())
            .build();
    }
    
    public static <T> ReactiveApiResponse<T> error(String message) {
        return ReactiveApiResponse.<T>builder()
            .success(false)
            .message(message)
            .data(null)
            .timestamp(LocalDateTime.now())
            .build();
    }
}

/**
 * 网站信息模型
 * 相当于Python的WebsiteInfo
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
class WebsiteInfo {
    
    private String url;
    
    private String title;
    
    private Integer statusCode;
    
    private Long responseTimeMs;
    
    private Integer contentLength;
    
    private Map<String, String> headers;
}

/**
 * 流式数据模型
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
class StreamData {
    
    private Integer number;
    
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss")
    private LocalDateTime timestamp;
    
    private Double progress;
    
    private Boolean finished;
}

/**
 * 日志条目模型
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
class LogEntry {
    
    private Long id;
    
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss")
    private LocalDateTime timestamp;
    
    private String level;
    
    private String message;
    
    private String module;
    
    private Double elapsed;
}

// ============================================================================
// 3. 响应式任务管理服务
// ============================================================================

/**
 * 响应式任务管理服务
 * 相当于Python的AsyncTaskManager
 * 
 * WebFlux特点：
 * 1. 使用Reactor的Mono和Flux
 * 2. 支持背压处理
 * 3. 声明式异步编程
 * 4. 完整的响应式流支持
 */
@Service
@Slf4j
class ReactiveTaskService {
    
    private final Map<String, ReactiveTaskResponse> tasks = new ConcurrentHashMap<>();
    private final AtomicLong taskIdGenerator = new AtomicLong(1);
    
    /**
     * 创建响应式任务
     */
    public Mono<String> createTask(ReactiveTaskRequest request) {
        return Mono.fromCallable(() -> {
            String taskId = "task_" + System.currentTimeMillis() + "_" + 
                           taskIdGenerator.getAndIncrement();
            
            ReactiveTaskResponse task = ReactiveTaskResponse.builder()
                .taskId(taskId)
                .name(request.getName())
                .status("pending")
                .progress(0.0)
                .createdAt(LocalDateTime.now())
                .build();
            
            tasks.put(taskId, task);
            
            // 启动异步任务
            startTask(taskId, request).subscribe();
            
            return taskId;
        })
        .subscribeOn(Schedulers.boundedElastic());
    }
    
    /**
     * 启动任务执行
     */
    private Mono<Void> startTask(String taskId, ReactiveTaskRequest request) {
        return Mono.defer(() -> {
            ReactiveTaskResponse task = tasks.get(taskId);
            task.setStatus("running");
            task.setStartedAt(LocalDateTime.now());
            
            switch (request.getTaskType()) {
                case "io_intensive":
                    return ioIntensiveTask(taskId, request);
                case "cpu_intensive":
                    return cpuIntensiveTask(taskId, request);
                case "mixed":
                    return mixedTask(taskId, request);
                default:
                    return defaultTask(taskId, request);
            }
        })
        .doOnError(error -> {
            ReactiveTaskResponse task = tasks.get(taskId);
            task.setStatus("failed");
            task.setError(error.getMessage());
            task.setCompletedAt(LocalDateTime.now());
            log.error("Task {} failed: {}", taskId, error.getMessage());
        })
        .onErrorResume(error -> Mono.empty());
    }
    
    /**
     * I/O密集型任务
     */
    private Mono<Void> ioIntensiveTask(String taskId, ReactiveTaskRequest request) {
        return Flux.range(1, 10)
            .delayElements(Duration.ofMillis(request.getDuration() * 100L))
            .flatMap(step -> {
                // 模拟HTTP请求
                return WebClient.create()
                    .get()
                    .uri("https://httpbin.org/delay/0.1")
                    .retrieve()
                    .bodyToMono(Map.class)
                    .map(response -> "Step " + step + ": " + response.get("url"))
                    .onErrorReturn("Step " + step + ": Error")
                    .doOnNext(result -> {
                        // 更新进度
                        ReactiveTaskResponse task = tasks.get(taskId);
                        task.setProgress(step * 10.0);
                        log.info("Task {} progress: {}%", taskId, step * 10);
                    });
            })
            .collectList()
            .doOnNext(results -> {
                // 任务完成
                ReactiveTaskResponse task = tasks.get(taskId);
                task.setStatus("completed");
                task.setProgress(100.0);
                task.setResult(Map.of(
                    "type", "io_intensive",
                    "steps_completed", results.size(),
                    "results", results,
                    "total_duration", request.getDuration()
                ));
                task.setCompletedAt(LocalDateTime.now());
            })
            .then();
    }
    
    /**
     * CPU密集型任务
     */
    private Mono<Void> cpuIntensiveTask(String taskId, ReactiveTaskRequest request) {
        return Mono.fromCallable(() -> {
            // CPU密集型计算（计算质数）
            long startTime = System.currentTimeMillis();
            List<Integer> primes = new ArrayList<>();
            int num = 2;
            
            while (System.currentTimeMillis() - startTime < request.getDuration() * 1000L) {
                boolean isPrime = true;
                for (int i = 2; i <= Math.sqrt(num); i++) {
                    if (num % i == 0) {
                        isPrime = false;
                        break;
                    }
                }
                
                if (isPrime) {
                    primes.add(num);
                }
                
                num++;
            }
            
            return Map.of(
                "primes_found", primes.size(),
                "largest_prime", primes.isEmpty() ? 0 : primes.get(primes.size() - 1),
                "actual_duration", (System.currentTimeMillis() - startTime) / 1000.0
            );
        })
        .subscribeOn(Schedulers.boundedElastic())
        .doOnNext(result -> {
            // 任务完成
            ReactiveTaskResponse task = tasks.get(taskId);
            task.setStatus("completed");
            task.setProgress(100.0);
            task.setResult(Map.of(
                "type", "cpu_intensive",
                "computation_result", result
            ));
            task.setCompletedAt(LocalDateTime.now());
        })
        .then();
    }
    
    /**
     * 混合型任务
     */
    private Mono<Void> mixedTask(String taskId, ReactiveTaskRequest request) {
        return Mono.defer(() -> {
            // 第一阶段：I/O操作
            Mono<List<String>> ioPhase = Flux.range(1, 3)
                .delayElements(Duration.ofSeconds(request.getDuration() / 6L))
                .map(i -> "IO Operation " + i + " completed")
                .collectList();
            
            // 第二阶段：CPU操作
            Mono<Integer> cpuPhase = Mono.fromCallable(() -> {
                return IntStream.range(1, 1000000)
                    .map(i -> i * i)
                    .sum();
            })
            .subscribeOn(Schedulers.boundedElastic());
            
            return Mono.zip(ioPhase, cpuPhase)
                .doOnNext(tuple -> {
                    ReactiveTaskResponse task = tasks.get(taskId);
                    task.setStatus("completed");
                    task.setProgress(100.0);
                    task.setResult(Map.of(
                        "type", "mixed",
                        "io_operations", tuple.getT1(),
                        "cpu_result", tuple.getT2(),
                        "total_duration", request.getDuration()
                    ));
                    task.setCompletedAt(LocalDateTime.now());
                })
                .then();
        });
    }
    
    /**
     * 默认任务
     */
    private Mono<Void> defaultTask(String taskId, ReactiveTaskRequest request) {
        return Flux.range(1, 5)
            .delayElements(Duration.ofSeconds(request.getDuration() / 5L))
            .doOnNext(step -> {
                ReactiveTaskResponse task = tasks.get(taskId);
                task.setProgress(step * 20.0);
            })
            .then()
            .doOnSuccess(v -> {
                ReactiveTaskResponse task = tasks.get(taskId);
                task.setStatus("completed");
                task.setProgress(100.0);
                task.setResult(Map.of(
                    "type", "default",
                    "message", "Task " + request.getName() + " completed successfully",
                    "duration", request.getDuration()
                ));
                task.setCompletedAt(LocalDateTime.now());
            });
    }
    
    /**
     * 获取任务状态
     */
    public Mono<ReactiveTaskResponse> getTask(String taskId) {
        return Mono.fromCallable(() -> tasks.get(taskId))
            .switchIfEmpty(Mono.error(new RuntimeException("Task not found: " + taskId)));
    }
    
    /**
     * 获取所有任务
     */
    public Flux<ReactiveTaskResponse> getAllTasks() {
        return Flux.fromIterable(tasks.values())
            .sort((t1, t2) -> t2.getCreatedAt().compareTo(t1.getCreatedAt()));
    }
    
    /**
     * 取消任务
     */
    public Mono<Boolean> cancelTask(String taskId) {
        return Mono.fromCallable(() -> {
            ReactiveTaskResponse task = tasks.get(taskId);
            if (task != null && "running".equals(task.getStatus())) {
                task.setStatus("cancelled");
                task.setCompletedAt(LocalDateTime.now());
                return true;
            }
            return false;
        });
    }
}

// ============================================================================
// 4. 响应式HTTP客户端服务
// ============================================================================

/**
 * 响应式HTTP客户端服务
 * 相当于Python的AsyncHttpService
 */
@Service
@Slf4j
class ReactiveHttpService {
    
    private final WebClient webClient;
    
    public ReactiveHttpService() {
        this.webClient = WebClient.builder()
            .codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(1024 * 1024))
            .build();
    }
    
    /**
     * 获取网站信息
     */
    public Mono<WebsiteInfo> fetchWebsiteInfo(String url) {
        long startTime = System.currentTimeMillis();
        
        return webClient.get()
            .uri(url)
            .retrieve()
            .toEntity(String.class)
            .map(response -> {
                long responseTime = System.currentTimeMillis() - startTime;
                String content = response.getBody();
                
                // 简单提取标题
                String title = "Unknown";
                if (content != null && content.contains("<title>")) {
                    int start = content.indexOf("<title>") + 7;
                    int end = content.indexOf("</title>", start);
                    if (end > start) {
                        title = content.substring(start, end).trim();
                    }
                }
                
                return WebsiteInfo.builder()
                    .url(url)
                    .title(title)
                    .statusCode(response.getStatusCodeValue())
                    .responseTimeMs(responseTime)
                    .contentLength(content != null ? content.length() : 0)
                    .headers(response.getHeaders().toSingleValueMap())
                    .build();
            })
            .onErrorReturn(WebsiteInfo.builder()
                .url(url)
                .title("Error")
                .statusCode(0)
                .responseTimeMs(System.currentTimeMillis() - startTime)
                .contentLength(0)
                .headers(new HashMap<>())
                .build()
            )
            .timeout(Duration.ofSeconds(10))
            .retry(2);
    }
    
    /**
     * 并发获取多个网站信息
     */
    public Flux<WebsiteInfo> fetchMultipleWebsites(List<String> urls) {
        return Flux.fromIterable(urls)
            .flatMap(this::fetchWebsiteInfo, 5) // 限制并发数为5
            .onErrorContinue((error, item) -> {
                log.warn("Failed to fetch website info for {}: {}", item, error.getMessage());
            });
    }
}

// ============================================================================
// 5. 响应式控制器
// ============================================================================

/**
 * 响应式Web控制器
 * 相当于Python的FastAPI路由
 * 
 * WebFlux特点：
 * 1. 返回Mono<T>或Flux<T>
 * 2. 支持背压处理
 * 3. 声明式异步编程
 * 4. 完整的响应式流支持
 */
@RestController
@RequestMapping("/reactive")
@Slf4j
class ReactiveController {
    
    private final ReactiveTaskService taskService;
    private final ReactiveHttpService httpService;
    
    public ReactiveController(ReactiveTaskService taskService, ReactiveHttpService httpService) {
        this.taskService = taskService;
        this.httpService = httpService;
    }
    
    /**
     * 响应式首页
     * 相当于Python的async def root()
     */
    @GetMapping("/")
    public Mono<Map<String, Object>> root() {
        return Mono.delay(Duration.ofMillis(100)) // 模拟异步操作
            .map(delay -> Map.of(
                "message", "欢迎使用SpringBoot WebFlux响应式编程示例",
                "features", Arrays.asList(
                    "响应式任务管理",
                    "并发HTTP请求",
                    "流式响应",
                    "背压处理",
                    "响应式数据流"
                ),
                "timestamp", LocalDateTime.now()
            ));
    }
    
    /**
     * 创建响应式任务
     */
    @PostMapping("/tasks")
    public Mono<ReactiveApiResponse<Map<String, Object>>> createTask(
            @Valid @RequestBody ReactiveTaskRequest request) {
        
        return taskService.createTask(request)
            .map(taskId -> {
                log.info("创建响应式任务: {} - {}", taskId, request.getName());
                
                return ReactiveApiResponse.success(
                    "响应式任务创建成功",
                    Map.of(
                        "task_id", taskId,
                        "status", "pending",
                        "estimated_duration", request.getDuration()
                    )
                );
            })
            .onErrorReturn(ReactiveApiResponse.error("创建任务失败"));
    }
    
    /**
     * 获取任务状态
     */
    @GetMapping("/tasks/{taskId}")
    public Mono<ReactiveApiResponse<ReactiveTaskResponse>> getTaskStatus(
            @PathVariable String taskId) {
        
        return taskService.getTask(taskId)
            .map(task -> ReactiveApiResponse.success("获取任务状态成功", task))
            .onErrorReturn(ReactiveApiResponse.error("任务不存在: " + taskId));
    }
    
    /**
     * 获取所有任务
     */
    @GetMapping("/tasks")
    public Mono<ReactiveApiResponse<List<ReactiveTaskResponse>>> getAllTasks() {
        return taskService.getAllTasks()
            .collectList()
            .map(tasks -> ReactiveApiResponse.success(
                "获取任务列表成功，共 " + tasks.size() + " 个任务",
                tasks
            ));
    }
    
    /**
     * 取消任务
     */
    @DeleteMapping("/tasks/{taskId}")
    public Mono<ReactiveApiResponse<Map<String, Object>>> cancelTask(
            @PathVariable String taskId) {
        
        return taskService.cancelTask(taskId)
            .map(success -> {
                if (success) {
                    log.info("取消响应式任务: {}", taskId);
                    return ReactiveApiResponse.success(
                        "任务取消成功",
                        Map.of("cancelled_task_id", taskId)
                    );
                } else {
                    return ReactiveApiResponse.error("任务不存在或已完成: " + taskId);
                }
            });
    }
    
    /**
     * 批量检查网站状态
     */
    @PostMapping("/websites/batch-check")
    public Mono<ReactiveApiResponse<Map<String, Object>>> batchCheckWebsites(
            @RequestBody List<String> urls) {
        
        if (urls.size() > 20) {
            return Mono.just(ReactiveApiResponse.error("最多支持20个URL"));
        }
        
        long startTime = System.currentTimeMillis();
        
        return httpService.fetchMultipleWebsites(urls)
            .collectList()
            .map(websites -> {
                long totalTime = System.currentTimeMillis() - startTime;
                
                log.info("批量检查 {} 个网站，耗时 {}ms", urls.size(), totalTime);
                
                return ReactiveApiResponse.success(
                    "批量检查完成，共检查 " + urls.size() + " 个网站",
                    Map.of(
                        "websites", websites,
                        "total_time_ms", totalTime,
                        "average_time_ms", urls.isEmpty() ? 0 : totalTime / urls.size()
                    )
                );
            });
    }
    
    /**
     * 流式响应示例
     * 相当于Python的StreamingResponse
     */
    @GetMapping(value = "/stream/numbers", produces = MediaType.TEXT_PLAIN_VALUE)
    public Flux<String> streamNumbers(
            @RequestParam(defaultValue = "10") int count,
            @RequestParam(defaultValue = "0.5") double delay) {
        
        if (count > 100) {
            return Flux.error(new IllegalArgumentException("最多支持100个数字"));
        }
        
        return Flux.range(1, count)
            .delayElements(Duration.ofMillis((long) (delay * 1000)))
            .map(i -> {
                StreamData data = StreamData.builder()
                    .number(i)
                    .timestamp(LocalDateTime.now())
                    .progress((double) i / count * 100)
                    .finished(i == count)
                    .build();
                
                return "data: " + data.toString() + "\n\n";
            })
            .concatWith(Mono.just("data: {\"finished\": true}\n\n"));
    }
    
    /**
     * 模拟日志流
     */
    @GetMapping(value = "/stream/logs", produces = MediaType.TEXT_PLAIN_VALUE)
    public Flux<String> streamLogs(
            @RequestParam(defaultValue = "10") int duration) {
        
        if (duration > 60) {
            return Flux.error(new IllegalArgumentException("最长支持60秒"));
        }
        
        AtomicLong logCount = new AtomicLong(0);
        long startTime = System.currentTimeMillis();
        
        return Flux.interval(Duration.ofMillis(100), Duration.ofSeconds(1))
            .take(Duration.ofSeconds(duration))
            .map(tick -> {
                long count = logCount.incrementAndGet();
                String[] logTypes = {"INFO", "DEBUG", "WARNING", "ERROR"};
                String logType = logTypes[(int) (count % logTypes.length)];
                
                LogEntry logEntry = LogEntry.builder()
                    .id(count)
                    .timestamp(LocalDateTime.now())
                    .level(logType)
                    .message("这是第 " + count + " 条 " + logType + " 日志")
                    .module("module_" + (count % 5 + 1))
                    .elapsed((System.currentTimeMillis() - startTime) / 1000.0)
                    .build();
                
                return "data: " + logEntry.toString() + "\n\n";
            })
            .concatWith(Mono.just("data: {\"finished\": true}\n\n"));
    }
    
    /**
     * 并发处理演示
     */
    @GetMapping("/concurrent/demo")
    public Mono<ReactiveApiResponse<Map<String, Object>>> concurrentDemo() {
        long startTime = System.currentTimeMillis();
        
        // 定义多个响应式任务
        Mono<String> task1 = Mono.delay(Duration.ofSeconds(1))
            .map(delay -> "任务1完成");
        
        Mono<String> task2 = Mono.delay(Duration.ofMillis(1500))
            .map(delay -> "任务2完成");
        
        Mono<String> task3 = Mono.delay(Duration.ofMillis(800))
            .map(delay -> "任务3完成");
        
        Mono<String> httpTask = WebClient.create()
            .get()
            .uri("https://httpbin.org/delay/1")
            .retrieve()
            .bodyToMono(Map.class)
            .map(response -> "HTTP任务完成: " + response.get("url"))
            .onErrorReturn("HTTP任务失败");
        
        // 并发执行所有任务
        return Mono.zip(task1, task2, task3, httpTask)
            .map(tuple -> {
                long totalTime = System.currentTimeMillis() - startTime;
                
                List<String> results = Arrays.asList(
                    tuple.getT1(),
                    tuple.getT2(),
                    tuple.getT3(),
                    tuple.getT4()
                );
                
                log.info("并发演示完成，耗时 {}ms", totalTime);
                
                return ReactiveApiResponse.success(
                    "并发演示完成",
                    Map.of(
                        "results", results,
                        "total_time_ms", totalTime,
                        "note", "如果串行执行，预计需要4300ms以上"
                    )
                );
            });
    }
    
    /**
     * 性能对比演示
     */
    @GetMapping("/performance/comparison")
    public Mono<ReactiveApiResponse<Map<String, Object>>> performanceComparison() {
        // 响应式版本
        Mono<Long> reactiveVersion = Mono.fromCallable(System::currentTimeMillis)
            .flatMap(startTime -> {
                return Flux.range(1, 10)
                    .flatMap(i -> Mono.delay(Duration.ofMillis(100)))
                    .then(Mono.fromCallable(() -> System.currentTimeMillis() - startTime));
            });
        
        // 同步版本（在弹性调度器中执行）
        Mono<Long> syncVersion = Mono.fromCallable(() -> {
            long startTime = System.currentTimeMillis();
            
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            
            return System.currentTimeMillis() - startTime;
        })
        .subscribeOn(Schedulers.boundedElastic());
        
        // 执行对比
        return Mono.zip(reactiveVersion, syncVersion)
            .map(tuple -> {
                long reactiveTime = tuple.getT1();
                long syncTime = tuple.getT2();
                double improvement = (double) (syncTime - reactiveTime) / syncTime * 100;
                
                return ReactiveApiResponse.success(
                    "性能对比完成",
                    Map.of(
                        "reactive_time_ms", reactiveTime,
                        "sync_time_ms", syncTime,
                        "improvement_percentage", improvement,
                        "note", "响应式版本在I/O密集型任务中表现更优"
                    )
                );
            });
    }
}

// ============================================================================
// 6. 配置类
// ============================================================================

/**
 * WebFlux配置
 */
@Configuration
@EnableAsync
class WebFluxConfig {
    
    /**
     * 配置WebClient
     */
    @Bean
    public WebClient webClient() {
        return WebClient.builder()
            .codecs(configurer -> {
                configurer.defaultCodecs().maxInMemorySize(1024 * 1024);
            })
            .build();
    }
    
    /**
     * 配置异步任务执行器
     */
    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("async-task-");
        executor.initialize();
        return executor;
    }
}

// ============================================================================
// 7. 全局异常处理
// ============================================================================

/**
 * 响应式全局异常处理器
 */
@ControllerAdvice
@Slf4j
class ReactiveExceptionHandler {
    
    /**
     * 处理非法参数异常
     */
    @ExceptionHandler(IllegalArgumentException.class)
    public Mono<ResponseEntity<ReactiveApiResponse<Object>>> handleIllegalArgument(
            IllegalArgumentException e) {
        
        log.warn("非法参数异常: {}", e.getMessage());
        
        return Mono.just(ResponseEntity.badRequest()
            .body(ReactiveApiResponse.error("参数错误: " + e.getMessage())));
    }
    
    /**
     * 处理通用异常
     */
    @ExceptionHandler(Exception.class)
    public Mono<ResponseEntity<ReactiveApiResponse<Object>>> handleGeneral(Exception e) {
        log.error("未处理异常: {} - {}", e.getClass().getSimpleName(), e.getMessage());
        
        return Mono.just(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
            .body(ReactiveApiResponse.error("内部服务器错误")));
    }
}

// ============================================================================
// 8. 应用程序主类
// ============================================================================

/**
 * SpringBoot WebFlux应用程序主类
 */
@SpringBootApplication
@Slf4j
public class SpringBootWebFluxApplication {
    
    public static void main(String[] args) {
        log.info("启动SpringBoot WebFlux响应式应用...");
        SpringApplication.run(SpringBootWebFluxApplication.class, args);
        log.info("SpringBoot WebFlux应用启动完成");
    }
}

// ============================================================================
// 9. 配置文件 application.yml
// ============================================================================

/*
# application.yml 配置文件内容

server:
  port: 8081
  netty:
    connection-timeout: 30s
    h2c-max-content-length: 1MB

spring:
  application:
    name: springboot-webflux-demo
  
  webflux:
    base-path: /
    static-path-pattern: /static/**
  
  # 响应式数据库配置（如果使用R2DBC）
  # r2dbc:
  #   url: r2dbc:h2:mem:///testdb
  #   username: sa
  #   password: 

# 日志配置
logging:
  level:
    com.example.webflux: INFO
    reactor.netty: DEBUG
    org.springframework.web.reactive: DEBUG
  pattern:
    console: "%d{yyyy-MM-dd HH:mm:ss} - %logger{36} - %level - %msg%n"

# 管理端点配置
management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics
  endpoint:
    health:
      show-details: always
*/

// ============================================================================
// 10. Maven依赖配置 pom.xml
// ============================================================================

/*
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.0</version>
        <relativePath/>
    </parent>
    
    <groupId>com.example</groupId>
    <artifactId>springboot-webflux-demo</artifactId>
    <version>1.0.0</version>
    <name>SpringBoot WebFlux Demo</name>
    <description>SpringBoot WebFlux响应式编程示例</description>
    
    <properties>
        <java.version>11</java.version>
    </properties>
    
    <dependencies>
        <!-- SpringBoot WebFlux Starter -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        
        <!-- SpringBoot Validation -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-validation</artifactId>
        </dependency>
        
        <!-- SpringBoot Actuator -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        
        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        
        <!-- Reactor Test -->
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
        
        <!-- SpringBoot Test -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>
*/

// ============================================================================
// 11. FastAPI vs SpringBoot WebFlux 异步编程对比总结
// ============================================================================

/*
FastAPI vs SpringBoot WebFlux 异步编程详细对比:

1. 编程模型:
   - FastAPI: async/await，直观简洁
   - WebFlux: Mono/Flux，函数式响应式编程

2. 学习曲线:
   - FastAPI: 相对简单，Python开发者容易上手
   - WebFlux: 较陡峭，需要理解响应式编程概念

3. 异步语法:
   - FastAPI: async def function(), await operation()
   - WebFlux: Mono.fromCallable(), .flatMap(), .map()

4. 并发处理:
   - FastAPI: asyncio.gather(), asyncio.create_task()
   - WebFlux: Mono.zip(), Flux.merge(), .flatMap()

5. 流式处理:
   - FastAPI: async generator, StreamingResponse
   - WebFlux: Flux<T>, 原生背压支持

6. 错误处理:
   - FastAPI: try/except, exception_handler
   - WebFlux: .onErrorReturn(), .onErrorResume()

7. 背压处理:
   - FastAPI: 需要手动控制
   - WebFlux: 内置背压机制

8. 性能特点:
   - FastAPI: 高性能，内存占用低
   - WebFlux: 高吞吐量，适合高并发场景

9. 生态系统:
   - FastAPI: aiohttp, asyncpg, motor等
   - WebFlux: Reactor, R2DBC, Spring生态

10. 调试难度:
    - FastAPI: 相对容易调试
    - WebFlux: 调试复杂，需要理解响应式流

11. 适用场景:
    - FastAPI: API服务，微服务，快速开发
    - WebFlux: 高并发系统，流式数据处理

12. 开发效率:
    - FastAPI: 开发速度快，代码简洁
    - WebFlux: 开发复杂，但功能强大

13. 企业级特性:
    - FastAPI: 轻量级，适合中小型项目
    - WebFlux: 企业级，适合大型复杂系统

14. 社区支持:
    - FastAPI: 快速增长的社区
    - WebFlux: 成熟的Spring社区支持

15. 总结:
    - FastAPI: 简单、快速、直观的异步编程
    - WebFlux: 强大、完整的响应式编程框架
*/